package com.github.slprj.slake.net;

import com.github.slprj.slake.common.AbstractLifeCircle;
import com.github.slprj.slake.common.SlThreadFactory;
import com.github.slprj.slake.common.SlTime;

import java.nio.ByteBuffer;
import java.util.concurrent.*;
import java.util.function.Function;

/**
 * 初始化网络处理相关的参数
 * -业务处理线程
 * -限流器
 *
 * @author YangXiaoLong
 * 2022/8/31 21:38
 */
public abstract class NioNet extends AbstractLifeCircle {
    private final NioConfig config;
    private final Semaphore semaphore;
    protected final NioStatus nioStatus = new NioStatus();
    protected ExecutorService bizExecutor;

    public NioNet(NioConfig config) {
        this.config = config;
        this.semaphore = new Semaphore(config.getMaxOutRequests());
        nioStatus.setRequestSemaphore(semaphore);
    }

    public void register(int cmd, ReqProcessor processor) {
        this.nioStatus.registerProcessor(cmd, processor);
    }

    protected <T> CompletableFuture<T> errorFuture(Throwable e) {
        CompletableFuture<T> f = new CompletableFuture<>();
        f.completeExceptionally(e);
        return f;
    }

    /**
     * 初始化业务线程池
     */
    protected void initBizExecutor() {
        if (config.getBizThreads() > 0) {
            bizExecutor = new ThreadPoolExecutor(config.getBizThreads(),
                    config.getBizThreads(),
                    1,
                    TimeUnit.MINUTES,
                    new ArrayBlockingQueue<>(config.getBizQueueSize()),
                    new SlThreadFactory(config.getName() + "Biz", false)
            );
            nioStatus.setBizExecutor(bizExecutor);
        }
    }

    protected void shutdownBizExecutor(SlTime timeout) {
        if (bizExecutor != null) {
            bizExecutor.shutdown();
            //当前线程已经被中断了
            if (Thread.currentThread().isInterrupted()) {
                return;
            }
            long rest = timeout.rest(TimeUnit.MILLISECONDS);
            if (rest > 0) {
                try {
                    //等待一定的时间,如果还没有执行完成,就直接关闭线程池
                    bizExecutor.awaitTermination(rest, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    ThreadUtils.restoreInterruptStatus();
                }
            }
        }
    }

    /**
     * 提供消息发送的能力
     * 向谁发送
     * 内容是什么
     * 如何解码
     * 超时时间多久
     */
    protected CompletableFuture<ReadFrame> sendRequest(NioWorker worker, Peer peer, WriteFrame request,
                                                       Decoder decoder, SlTime timeout) {
        boolean acquire = false;
        try {
            if (status != LifeStatus.running) {
                return errorFuture(new IllegalStateException("error state: " + status));
            }

            acquire = this.semaphore.tryAcquire(timeout.rest(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
            if (acquire) {
                CompletableFuture<ReadFrame> future = new CompletableFuture<>();
                worker.writeReqInBizThreads(peer, request, decoder, timeout, future);
                return future.thenApply(new RespConvertor(decoder));
            } else {
                return errorFuture(new NetTimeoutException(
                        "too many pending requests, client wait permit timeout in "
                                + timeout.getTimeout(TimeUnit.MILLISECONDS) + " ms"));
            }
        } catch (InterruptedException e) {
            if (acquire) {
                this.semaphore.release();
            }
            return errorFuture(e);
        } catch (Throwable e) {
            if (acquire) {
                this.semaphore.release();
            }
            return errorFuture(new NetException("submit task error", e));
        }
    }

    private static class RespConvertor implements Function<ReadFrame, ReadFrame> {
        private Decoder decoder;
        RespConvertor(Decoder decoder) {
            this.decoder = decoder;
        }
        @Override
        public ReadFrame apply(ReadFrame frame) {
            if (frame.getRespCode() != CmdCodes.SUCCESS) {
                throw new NetCodeException(frame.getRespCode());
            }
            if (!decoder.decodeInIoThread()) {
                ByteBuffer buf = (ByteBuffer) frame.getBody();
                Object body = decoder.decode(buf);
                frame.setBody(body);
            }
            return frame;
        }
    }

}
